/*-
 * Copyright 2012 Guram Dukashvili
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */
//---------------------------------------------------------------------------
#include "sys.h"
#include "str.h"
#include "idx.h"
//---------------------------------------------------------------------------
static const struct signature index_file_header_signature = {
  {
    0x6E, 0x77, 0xBF, 0xB6, 0xD3, 0x86, 0xD9, 0xD0,
    0xAA, 0xAE, 0xBB, 0xB9, 0xE6, 0xAD, 0x2F, 0x97,
    0x6F, 0x84, 0x01, 0x15, 0x4B, 0xC2, 0x16, 0x49,
    0x4E, 0xDA, 0x1C, 0xA7, 0x55, 0xD7, 0x34, 0x1E
  }
};
//---------------------------------------------------------------------------
struct index * index_new(void)
{
  struct index * idx = salloc(sizeof(struct index),NULL);
  if( idx != NULL )
    index_init(idx);
  return idx;
}
//---------------------------------------------------------------------------
static __inline__ intptr_t index_compare(
  const struct index * p1,const struct index * p2)
{
  intptr_t c = p1->id > p2->id ? 1 : p1->id < p2->id ? -1 : 0;
  if( c == 0 )
    c = p1->mvcc_id > p2->mvcc_id ? 1 : p1->mvcc_id < p2->mvcc_id ? -1 : 0;
  return c;
}
//---------------------------------------------------------------------------
static uint32_t index_link_hash(const struct index_link * data)
{
  return 
    hashf_u32(data->idx->id,
      hashf_u32(data->idx->mvcc_id,
        bsint_hash(&data->block_index,data->idx->block_index_size)));
}
//---------------------------------------------------------------------------
static __inline__ intptr_t index_link_compare(
  const struct index_link * p1,const struct index_link * p2)
{
  intptr_t c = index_compare(p1->idx,p2->idx);
  if( c == 0 )
    c = memcmp(&p1->block_index,&p2->block_index,p1->idx->block_index_size);
  return c;
}
//---------------------------------------------------------------------------
static __inline__ intptr_t index_link_change_compare(
  const struct index_link_change * p1,const struct index_link_change * p2)
{
  return index_link_compare(p1->link,p2->link);
}
//---------------------------------------------------------------------------
static __inline__ intptr_t index_blocks_map_compare(
  const struct index_block_mapper_entry * p1,
  const struct index_block_mapper_entry * p2,
  struct index * idx)
{
  return memcmp(&p1->block_index,&p2->block_index,idx->block_index_size);
}
//---------------------------------------------------------------------------
static __inline__ void jrn_file_handles_array_init(array * ar)
{
  array_init(ar);
  array_set_element_size(ar,sizeof(int),sizeof(int) * 8);
}
//---------------------------------------------------------------------------
static void index_disable_copy(void)
{
  errno = ENOSYS;
  perror(NULL);
#if COMPILE_RUNTIME_CHECKS
  fprintf(stderr,"undefined behavior in %s, %s, line %d\n",__PRETTY_FUNCTION__,__FILE__,__LINE__);
#endif
  abort();
}
//---------------------------------------------------------------------------
static struct shared_cache cache_runtime;
//---------------------------------------------------------------------------
void index_init_runtime(void)
{
  shared_cache_init(&cache_runtime);
}
//---------------------------------------------------------------------------
void index_destroy_runtime(void)
{
  shared_cache_destroy(&cache_runtime);
}
//---------------------------------------------------------------------------
void index_init(struct index * idx)
{
  idx->id = 0;
  idx->mvcc_id = 0;
  idx->key_size = 0;
  idx->data_size = 0;
  idx->node_entry_size = 0;
  idx->leaf_entry_size = 0;
  idx->block_index_size = 0;
  idx->node_max_key_count = 0;
  idx->leaf_max_key_count = 0;

  idx->cache = &cache_runtime;

  rwlock_init(&idx->readers_lock);

  array_init(&idx->changed_links);
  array_set_element_size(
    &idx->changed_links,
    sizeof(struct index_link_change),
    sizeof(idx->changed_links_static_data)
  );
  idx->changed_links.element_compare = index_link_change_compare;

  rwlock_init(&idx->changed_links_lock);

  array_init(&idx->idx_file_handles);
  array_set_element_size(&idx->idx_file_handles,sizeof(int),
    sizeof(idx->idx_file_handles_static_data));

  array_init(&idx->jrn_file_handles);
  array_set_element_size(&idx->jrn_file_handles,sizeof(array)
    + sizeof(int) * 8, // add space for static data
  0);
  idx->jrn_file_handles.element_init = jrn_file_handles_array_init;
  idx->jrn_file_handles.element_copy = array_copy;
  idx->jrn_file_handles.element_destroy = array_destroy;

  mutex_init(&idx->file_handles_mutex);

  array_init(&idx->blocks_map);
  array_set_element_size(
    &idx->blocks_map,
    sizeof(struct index_block_mapper_entry),
    sizeof(idx->blocks_map_static_data)
  );
  idx->blocks_map.element_compare = index_blocks_map_compare;
  idx->blocks_map.object = idx;

  rwlock_init(&idx->blocks_map_lock);

  string_init(&idx->data_dir);

  thread_init(&idx->write_journal_thread);
  idx->write_journal_thread.entry = index_write_journal_entry;
  idx->write_journal_thread.object = idx;
  semaphore_init(&idx->write_journal_wait);

  semaphore_init_v(&idx->mvcc_lock,1);
  semaphore_init(&idx->mvcc_wait);
  vector_init(&idx->mvcc_indexes_pool);
  idx->mvcc_indexes_pool.element_new = index_new;
  idx->mvcc_indexes_pool.element_copy = index_disable_copy;
  idx->mvcc_indexes_pool.element_delete = index_delete;
  idx->mvcc_indexes_pool.element_compare = index_compare;
  thread_pool_init(&idx->mvcc_thread_pool);
  idx->mvcc_thread_pool.max_size = system_cores() + 1;
  idx->mvcc_pool_size = 4;
  idx->mvcc_ref_count = 0;
  idx->mvcc_count = 0;
  idx->mvcc_waiters = 0;
  idx->mvcc_inprogress = false;
}
//---------------------------------------------------------------------------
void index_delete(struct index * idx)
{
  index_destroy(idx);
  sfree(idx);
}
//---------------------------------------------------------------------------
void index_destroy(struct index * idx)
{
  intptr_t i;
  struct list_node * node;

  // perform shutdown
  thread_pool_destroy(&idx->mvcc_thread_pool);
  vector_destroy(&idx->mvcc_indexes_pool);

  idx->write_journal_thread.terminate = true;
  semaphore_post(&idx->write_journal_wait);
  thread_destroy(&idx->write_journal_thread);

  index_write_journal(idx);
  index_flush_journal(idx);

#if COMPILE_RUNTIME_CHECKS
  if( idx->blocks_map.count != 0 ){
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
  if( idx->changed_links.count != 0 ){
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
//  if( idx->links_linked.count != 0 ){
//    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
//    abort();
//  }
#endif

  node = idx->cache->links_linked.tail;

  while( node != NULL ){
    struct index_link * lnk = list_node2data(&idx->cache->links_linked,node);

#if COMPILE_RUNTIME_CHECKS
    if( lnk->idx == idx ){
      uint64_t block_index = bsint_get_be64toh(&lnk->block_index,lnk->idx->block_index_size);
      uint64_t block_extent = block_index / idx->cache->extent_size;
      uint64_t extent_block_index = block_extent == 0 ? 2 : block_extent * idx->cache->extent_size;
      if( lnk->linked != 0 ){
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
        abort();
      }
      if( extent_block_index != block_index ){
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
        abort();
      }
      if( lnk->list != &lnk->idx->cache->links_linked ){
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
        abort();
      }
    }
#endif
    node = node->prev;

    if( lnk->idx == idx ){
      list_remove(&lnk->idx->cache->links_linked,lnk);
      list_insert_head(&lnk->idx->cache->links_lru,lnk);
      lnk->list = &lnk->idx->cache->links_lru;
    }
  }

  node = idx->cache->links_lru.tail;

  while( node != NULL ){
    struct index_link * lnk = list_node2data(&idx->cache->links_lru,node);

    node = node->prev;

    if( lnk->idx == idx ){
      list_remove(&idx->cache->links_lru,lnk);
      hash_remove(&idx->cache->links_hash,lnk);
      rwlock_destroy(&lnk->lock);
#if COMPILE_RUNTIME_CHECKS
      if( lnk->linked != 0 ){
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
        abort();
      }
      if( lnk->list != &idx->cache->links_lru ){
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
        abort();
      }
#endif
      sfree(lnk);
    }
  }

  // call destructors

  semaphore_destroy(&idx->mvcc_wait);
  semaphore_destroy(&idx->mvcc_lock);

  semaphore_destroy(&idx->write_journal_wait);

  string_destroy(&idx->data_dir);

  array_destroy(&idx->blocks_map);

  for( i = idx->jrn_file_handles.count - 1; i >= 0; i-- )
    index_close_file_handles(idx,INDEX_JRN_FILE_HANDLE,i);
  array_destroy(&idx->jrn_file_handles);

  index_close_file_handles(idx,INDEX_IDX_FILE_HANDLE,0);
  array_destroy(&idx->idx_file_handles);

  rwlock_destroy(&idx->blocks_map_lock);
  mutex_destroy(&idx->file_handles_mutex);
  rwlock_destroy(&idx->changed_links_lock);

  rwlock_destroy(&idx->readers_lock);
}
//---------------------------------------------------------------------------
void index_drop(struct index * idx)
{
  intptr_t i;
  string filter;
  array dir_list;

  string_init(&filter);

  index_get_file_filter(idx,INDEX_JRN_FILE_HANDLE,&filter,idx->mvcc_id);

  dir_list_init(&dir_list);
  dir_list_get(&dir_list,filter.s,false,true);

  index_get_file_filter(idx,INDEX_IDX_FILE_HANDLE,&filter,idx->mvcc_id);
  dir_list_get(&dir_list,filter.s,false,true);

  for( i = idx->mvcc_pool_size - 1; i >= 0; i-- ){
    index_get_file_filter(idx,INDEX_JRN_FILE_HANDLE,&filter,i + 1);
    dir_list_get(&dir_list,filter.s,false,true);
    index_get_file_filter(idx,INDEX_IDX_FILE_HANDLE,&filter,i + 1);
    dir_list_get(&dir_list,filter.s,false,true);
  }

  for( i = dir_list.count - 1; i >= 0; i-- ){
    string * name = array_get(&dir_list,i);
    portable_unlink(name->s);
  }

  array_destroy(&dir_list);
  string_destroy(&filter);
}
//---------------------------------------------------------------------------
void index_set_key_size(struct index * idx,uintptr_t key_size)
{
  uintptr_t block_size;

  idx->key_size = key_size;

  idx->node_entry_size = idx->key_size + idx->block_index_size;
  block_size = ge2(idx->node_entry_size * 16 + sizeof(struct index_block));
  if( idx->cache->block_size < block_size ) idx->cache->block_size = (uint32_t) block_size;

  idx->leaf_entry_size = idx->key_size + idx->data_size;
  block_size = ge2(idx->leaf_entry_size * 16 + sizeof(struct index_block));
  if( idx->cache->block_size < block_size ) idx->cache->block_size = (uint32_t) block_size;

  idx->node_max_key_count =
    (idx->cache->block_size - sizeof(struct index_block)) / idx->node_entry_size;

  idx->leaf_max_key_count =
    (idx->cache->block_size - sizeof(struct index_block)) / idx->leaf_entry_size;
}
//---------------------------------------------------------------------------
void index_set_data_size(struct index * idx,uintptr_t data_size)
{
  idx->data_size = data_size;
  index_set_key_size(idx,idx->key_size);
}
//---------------------------------------------------------------------------
void index_set_block_index_size(struct index * idx,uintptr_t block_index_size)
{
  idx->block_index_size = block_index_size;
  index_set_key_size(idx,idx->key_size);
}
//---------------------------------------------------------------------------
void index_get_file_filter(struct index * idx,enum INDEX_FILE_TYPE type,string * filter,uintptr_t mvcc_id)
{
  if( type == INDEX_IDX_FILE_HANDLE ){
    if( mvcc_id == 0 )
      string_print(filter,"%s" PATH_DELIM "IDX-%06" PRIX32,idx->data_dir.s,idx->id);
    else
      string_print(filter,"%s" PATH_DELIM "IDX-%06" PRIX32 "-V%04" PRIX32,idx->data_dir.s,idx->id,mvcc_id - 1);
  }
  else if( type == INDEX_JRN_FILE_HANDLE ){
    if( mvcc_id == 0 )
      string_print(filter,"%s" PATH_DELIM "IDX-%06" PRIX32 "-JRN-" "????",
        idx->data_dir.s,idx->id);
    else
      string_print(filter,"%s" PATH_DELIM "IDX-%06" PRIX32 "-V%04" PRIX32 "-JRN-" "????",
        idx->data_dir.s,idx->id,mvcc_id - 1);
  }
#if COMPILE_RUNTIME_CHECKS
  else {
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
#endif
}
//---------------------------------------------------------------------------
void index_get_file_name(struct index * idx,enum INDEX_FILE_TYPE type,uintptr_t file_index,string * name)
{
  if( type == INDEX_IDX_FILE_HANDLE ){
    if( idx->mvcc_id == 0 )
      string_print(name,"%s" PATH_DELIM "IDX-%06" PRIX32,idx->data_dir.s,idx->id);
    else
      string_print(name,"%s" PATH_DELIM "IDX-%06" PRIX32 "-V%04" PRIX32,idx->data_dir.s,idx->id,idx->mvcc_id - 1);
  }
  else if( type == INDEX_JRN_FILE_HANDLE ){
    if( idx->mvcc_id == 0 )
      string_print(name,"%s" PATH_DELIM "IDX-%06" PRIX32 "-JRN-" "%04" PRIXPTR,
        idx->data_dir.s,idx->id,file_index);
    else
      string_print(name,"%s" PATH_DELIM "IDX-%06" PRIX32 "-V%04" PRIX32 "-JRN-" "%04" PRIXPTR,
        idx->data_dir.s,idx->id,idx->mvcc_id - 1,file_index);
  }
#if COMPILE_RUNTIME_CHECKS
  else {
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
#endif
}
//---------------------------------------------------------------------------
int index_get_file_handle(struct index * idx,enum INDEX_FILE_TYPE type,uintptr_t file_index,bool create)
{
  int file_handle = -1, flags = 0;

  errno = 0;

  mutex_acquire(&idx->file_handles_mutex);

  if( type == INDEX_IDX_FILE_HANDLE ){

    if( idx->idx_file_handles.count > 0 ){
      file_handle = *(int *) array_get(&idx->idx_file_handles,idx->idx_file_handles.count - 1);
      array_remove(&idx->idx_file_handles,idx->idx_file_handles.count - 1);
    }

    flags = O_RANDOM;
  }
  else if( type == INDEX_JRN_FILE_HANDLE ){

    if( file_index >= idx->jrn_file_handles.count )
      array_resize(&idx->jrn_file_handles,file_index + 1);

    if( errno == 0 ){
      array * handle_array = array_get(&idx->jrn_file_handles,file_index);

      if( handle_array->count > 0 ){
        file_handle = *(int *) array_get(handle_array,handle_array->count - 1);
        array_remove(handle_array,handle_array->count - 1);
      }

      //flags = O_DIRECT | O_FSYNC | O_SYNC;
    }
  }

  mutex_release(&idx->file_handles_mutex);

  if( file_handle < 0 ){
    string name;
    string_init(&name);

    index_get_file_name(idx,type,file_index,&name);

    file_handle = portable_open(
      name.s,
      (create ? O_CREAT : 0) | O_RDWR | O_BINARY | O_LARGEFILE | flags,
      S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP,
      true
    );

    string_destroy(&name);
  }

  return file_handle;
}
//---------------------------------------------------------------------------
void index_put_file_handle(struct index * idx,enum INDEX_FILE_TYPE type,uintptr_t file_index,int file_handle)
{
  mutex_acquire(&idx->file_handles_mutex);

  if( type == INDEX_IDX_FILE_HANDLE ){
    array_lookup_insert_copy(&idx->idx_file_handles,&file_handle);
#if COMPILE_RUNTIME_CHECKS
    if( errno == EEXIST ){
      fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
      abort();
    }
#endif
  }
  else if( type == INDEX_JRN_FILE_HANDLE ){
    errno = 0;

    if( file_index >= idx->jrn_file_handles.count )
      array_resize(&idx->jrn_file_handles,file_index + 1);

    if( errno == 0 ){
      array * handle_array = array_get(&idx->jrn_file_handles,file_index);
      array_lookup_insert_copy(handle_array,&file_handle);
#if COMPILE_RUNTIME_CHECKS
      if( errno == EEXIST ){
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
        abort();
      }
#endif
    }
    if( errno != 0 )
      portable_close(file_handle);
  }
#if COMPILE_RUNTIME_CHECKS
  else {
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
#endif

  mutex_release(&idx->file_handles_mutex);
}
//---------------------------------------------------------------------------
void index_close_file_handles(struct index * idx,enum INDEX_FILE_TYPE type,uintptr_t file_index)
{
  uintptr_t i;

  mutex_acquire(&idx->file_handles_mutex);

  if( type == INDEX_IDX_FILE_HANDLE ){
    for( i = 0; i < idx->idx_file_handles.count; i++ )
      portable_close(*(int *) array_get(&idx->idx_file_handles,i));
    array_resize(&idx->idx_file_handles,0);
  }
  else if( type == INDEX_JRN_FILE_HANDLE ){
    if( file_index < idx->jrn_file_handles.count ){
      array * handle_array = array_get(&idx->jrn_file_handles,file_index);

      for( i = 0; i < handle_array->count; i++ ){
        int * handle = array_get(handle_array,i);
        portable_close(*handle);
      }

      array_resize(handle_array,0);
    }
  }
#if COMPILE_RUNTIME_CHECKS
  else {
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
#endif

  mutex_release(&idx->file_handles_mutex);
}
//---------------------------------------------------------------------------
struct index_btree_stack * index_btree_stack_new(void)
{
  struct index_btree_stack * s = salloc(sizeof(struct index_btree_stack),NULL);
  if( s != NULL )
    index_btree_stack_init(s);
  return s;
}
//---------------------------------------------------------------------------
void index_btree_stack_init(struct index_btree_stack * s)
{
  s->sp = -1;
  array_init(&s->entries);
  array_set_element_size(
    &s->entries,
    sizeof(struct index_btree_stack_entry),
    sizeof(s->entries_static_data)
  );
  s->entries.element_init = index_btree_stack_entry_init;
}
//---------------------------------------------------------------------------
void index_btree_stack_destroy(struct index_btree_stack * s)
{
  array_destroy(&s->entries);
}
//---------------------------------------------------------------------------
void index_btree_stack_delete(struct index_btree_stack * s)
{
  index_btree_stack_destroy(s);
  sfree(s);
}
//---------------------------------------------------------------------------
void index_btree_stack_entry_init(struct index_btree_stack_entry * e)
{
  intptr_t i;
  for( i = sizeof(e->links) / sizeof(e->links[0]) - 1; i >= 0; i-- ){
    e->links[i] = NULL;
    e->links_change[i] = ILC_NONE;
  }
}
//---------------------------------------------------------------------------
struct shared_cache * shared_cache_new(void)
{
  struct shared_cache * cache = salloc(sizeof(struct shared_cache),NULL);
  if( cache != NULL )
    shared_cache_init(cache);
  return cache;
}
//---------------------------------------------------------------------------
void shared_cache_init(struct shared_cache * cache)
{
  struct index_link * link = NULL;

  cache->block_size = 4096;
  cache->extent_size = cache->block_size * UINT64_C(8);

  hash_init(&cache->links_hash);
  cache->links_hash.element_hash = index_link_hash;
  cache->links_hash.element_compare = index_link_compare;
  cache->links_hash.node = (uintptr_t) &link->hash_link - (uintptr_t) link;

  list_init(&cache->links_linked);
  cache->links_linked.node = (uintptr_t) &link->list_link - (uintptr_t) link;

  list_init(&cache->links_lru);
  cache->links_lru.node = (uintptr_t) &link->list_link - (uintptr_t) link;

  mutex_init(&cache->links_mutex);

  cache->max_size = 4 * 1024 * 1024;
  cache->cur_size = 0;
  cache->hit_count = 0;
  cache->miss_count = 0;
  cache->iowait_time = 0;

  cache->high_watermark = 0.999;
  cache->low_watermark = 0.921;
  cache->flush_watermark = 5. / 8.; // golden section
}
//---------------------------------------------------------------------------
void shared_cache_destroy(struct shared_cache * cache)
{
  list_destroy(&cache->links_lru);
  list_destroy(&cache->links_linked);
  hash_destroy(&cache->links_hash);
  mutex_destroy(&cache->links_mutex);
}
//---------------------------------------------------------------------------
void shared_cache_delete(struct shared_cache * cache)
{
  shared_cache_destroy(cache);
  sfree(cache);
}
//---------------------------------------------------------------------------
